package com.ruoyi.teaching.service.impl;

import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.stream.Collectors;

import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.core.toolkit.ObjectUtils;
import com.ruoyi.common.core.exception.ServiceException;
import com.ruoyi.common.core.utils.DateUtils;
import com.ruoyi.teaching.bo.HighInterviewMasteryListBo;
import com.ruoyi.teaching.bo.QueryInterviewMasterBo;
import com.ruoyi.teaching.bo.RelatedInterviewBo;
import com.ruoyi.teaching.flink.BatchESSink;
import com.ruoyi.teaching.flink.MySQLSource;
import com.ruoyi.teaching.index.Interview;
import com.ruoyi.teaching.utils.IKAnalyzerUtils;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.elasticsearch.index.query.QueryBuilders;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.stereotype.Service;
import com.ruoyi.teaching.mapper.YaHighInterviewMapper;
import com.ruoyi.teaching.domain.YaHighInterview;
import com.ruoyi.teaching.service.IYaHighInterviewService;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;

/**
 * 高频面试题Service业务层处理
 * 
 * @author ruoyi
 * @date 2025-03-14
 */
@Service
public class YaHighInterviewServiceImpl implements IYaHighInterviewService 
{
    @Autowired
    private YaHighInterviewMapper yaHighInterviewMapper;
    @Autowired
    private ElasticsearchRestTemplate elasticsearchRestTemplate;

    /**
     * 查询高频面试题
     * 
     * @param id 高频面试题主键
     * @return 高频面试题
     */
    @Override
    public YaHighInterview selectYaHighInterviewById(Long id)
    {
        return yaHighInterviewMapper.selectYaHighInterviewById(id);
    }

    /**
     * 查询高频面试题列表
     * 
     * @param yaHighInterview 高频面试题
     * @return 高频面试题
     */
    @Override
    public List<YaHighInterview> selectYaHighInterviewList(YaHighInterview yaHighInterview)
    {
        return yaHighInterviewMapper.selectYaHighInterviewList(yaHighInterview);
    }

    /**
     * 新增高频面试题
     * 
     * @param yaHighInterview 高频面试题
     * @return 结果
     */
    @Override
    public int insertYaHighInterview(YaHighInterview yaHighInterview)
    {
        yaHighInterview.setCreateTime(DateUtils.getNowDate());
        return yaHighInterviewMapper.insertYaHighInterview(yaHighInterview);
    }

    /**
     * 修改高频面试题
     * 
     * @param yaHighInterview 高频面试题
     * @return 结果
     */
    @Override
    public int updateYaHighInterview(YaHighInterview yaHighInterview)
    {
        yaHighInterview.setUpdateTime(DateUtils.getNowDate());
        return yaHighInterviewMapper.updateYaHighInterview(yaHighInterview);
    }

    /**
     * 批量删除高频面试题
     * 
     * @param ids 需要删除的高频面试题主键
     * @return 结果
     */
    @Override
    public int deleteYaHighInterviewByIds(Long[] ids)
    {
        return yaHighInterviewMapper.deleteYaHighInterviewByIds(ids);
    }

    /**
     * 删除高频面试题信息
     * 
     * @param id 高频面试题主键
     * @return 结果
     */
    @Override
    public int deleteYaHighInterviewById(Long id)
    {
        return yaHighInterviewMapper.deleteYaHighInterviewById(id);
    }

    @Override
    @Transactional(readOnly = true)
    public List<HighInterviewMasteryListBo> selectYaInterviewMasteryList(QueryInterviewMasterBo queryInterviewMasterBo) {
        return yaHighInterviewMapper.selectYaInterviewMasteryList(queryInterviewMasterBo);
    }

    @Override
    @Transactional(readOnly = true)
    public RelatedInterviewBo queryRelatedInterview(Long courseDesignId, String courseContext) {
        /*RelatedInterviewBo result = RelatedInterviewBo.builder().build();
        //parse the sentence,and get the keywords
        Set<String> keywords = IKAnalyzerUtils.parseSentence(courseContext,false);
        result.setKeywords(keywords);
        List<YaHighInterview> interviewList = null;
        //if courseDesignId is not null
        if (ObjectUtils.isNotEmpty(courseDesignId)) {
            List<Long> interviewIds = yaHighInterviewMapper.selectHighInterviewByCourseDesignId(courseDesignId);
            if(CollectionUtils.isNotEmpty(interviewIds)) {
                result.setRelatedInterviewIdList(interviewIds);
                interviewList = yaHighInterviewMapper.queryRelatedInterviewByIds(interviewIds);
            }else{
                //query related interview by keywords
                interviewList = yaHighInterviewMapper.queryRelatedInterview(keywords);
            }
            result.setHighInterviewList(interviewList);
        }
        return result;*/
        RelatedInterviewBo result = new RelatedInterviewBo();
        Set<String> keywords = IKAnalyzerUtils.parseSentence(courseContext, false);
        result.setKeywords(keywords);

        List<Long> relatedInterviewIdList = yaHighInterviewMapper.selectHighInterviewByCourseDesignId(courseDesignId);
        result.setRelatedInterviewIdList(relatedInterviewIdList);

        //使用elasticsearchRestTemplate查询面试题Interview.class,根据question分词查询
        NativeSearchQueryBuilder nativeSearchQueryBuilder = new NativeSearchQueryBuilder();
        nativeSearchQueryBuilder.withQuery(QueryBuilders.matchQuery("question", courseContext));
        List<YaHighInterview> yaHighInterviewList = elasticsearchRestTemplate.search(nativeSearchQueryBuilder.build(), Interview.class)
                .getSearchHits()
                .stream()
                .map(searchHit -> {
                    Interview interview = searchHit.getContent();
                    YaHighInterview target = new YaHighInterview();
                    BeanUtils.copyProperties(interview, target);
                    return target;
                }).collect(Collectors.toList());

        result.setHighInterviewList(yaHighInterviewList);

        return result;
    }

    @Override
    @Transactional(readOnly = true)
    public List<YaHighInterview> queryRelatedInterviewByKeywords(List<String> keywords) {
        if (CollectionUtils.isEmpty(keywords)) {
            return Collections.emptyList();
        }
        return yaHighInterviewMapper.queryRelatedInterview(new HashSet<>(keywords));
    }


    // 查询导出进度
    @Override
    public Integer queryExportEsProgress() {
        return MySQLSource.getPercentage();
    }

    // 导出数据到ES
    // 全局锁(1个信号量)
    public final Semaphore lock = new Semaphore(1);
    // 创建一个单线程的线程池,使用线程池执行异步任务,多线程优化响应速度
    private static final ExecutorService SINGLE_THREAD_POOL = Executors.newSingleThreadExecutor();
    @Override
    public void exportEs() {
        boolean b = lock.tryAcquire(1);
        if (!b) {
            throw new ServiceException("其他线程正在数据同步");
        }
        SINGLE_THREAD_POOL.execute(() -> {
            try {
                // 获取 Flink 的流处理执行环境
                // StreamExecutionEnvironment 是 Flink 流处理程序的上下文环境，用于定义和执行数据流任务
                StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

                BatchESSink batchESSink = new BatchESSink();
                MySQLSource mySQLSource = new MySQLSource();
                // 设置并行度为 1
                // 并行度决定了 Flink 作业中每个算子的并发执行数量，这里设置为 1 表示单线程执行
                env.setParallelism(1);

                // 添加数据源（Source）
                // 使用 MySQLSource 作为数据源，从 MySQL 数据库中读取 YaHighInterview 数据流
                DataStreamSource<YaHighInterview> source = env.addSource(mySQLSource);

                // 添加数据输出（Sink）
                // 将从 MySQL 读取的数据流通过 BatchESSink 输出到 Elasticsearch 中
                source.addSink(batchESSink);

                // 执行 Flink 作业
                // 调用 execute 方法启动 Flink 流处理作业，作业名称为 "MySQL to ES Export with Flink"
                env.execute("MySQL to ES Export with Flink");
            } catch (Exception e) {
                // 捕获异常并抛出运行时异常
                throw new ServiceException("全量导出到es异常");
            }finally {
                //当任务执行完后才能解锁
                lock.release();
            }
        });
    }

    @PostConstruct
    public void init(){

    }
}
